package com.wang.lottery.application.mp.consumer;

import cn.hutool.json.JSONUtil;
import com.wang.lottery.application.mp.producer.KafkaProducer;
import com.wang.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
import com.wang.lottery.domain.activity.repository.IActivityRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * @author wjh
 * @create 2023-11-16 16:43
 * 监听活动库存扣减记录，同步到数据库
 */
@Component
@Slf4j
public class LotteryActivityPartakeRecordListener {
    @Resource
    private IActivityRepository activityRepository;
    @KafkaListener(topics = KafkaProducer.TOPIC_ACTIVITY_PARTAKE,groupId = KafkaProducer.TOPIC_LOTTERY_GROUP)
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> optional = Optional.ofNullable(record.value());
        if (optional.isPresent()) {
            log.info("收到消息：{}", optional.get());
            Object message = optional.get();
            ActivityPartakeRecordVO activityPartakeRecordVO = JSONUtil.toBean(JSONUtil.parseObj(message), ActivityPartakeRecordVO.class);
            activityRepository.updateActivityStock(activityPartakeRecordVO);
            //消费完成
            ack.acknowledge();
        }
    }
}
